package com.atguigu.PracticalProject.app.dwd.db;

import com.atguigu.PracticalProject.app.BaseSqlApp;
import com.atguigu.PracticalProject.common.Constant;
import com.atguigu.PracticalProject.util.SQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableResult;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;
public class DwdTradeOrderDetail extends BaseSqlApp {

    public static void main(String[] args) {
        new DwdTradeOrderDetail().init(
                3007,
                3,
                "Edu_Dwd_DwdTradeOrderDetail"
        );
    }

    @Override
    protected void handle(StreamExecutionEnvironment env,
                          StreamTableEnvironment tEnv) {


        tEnv.getConfig().setIdleStateRetention(Duration.ofMinutes(30));
        //读取ods_db
        readOdsDb(tEnv,"Edu_Dwd_DwdTradeOrderDetail");

        //2order_detail
        Table orderDetail = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['course_id'] course_id, " +
                "data['course_name'] course_name, " +
                "data['order_id'] order_id, " +
                "data['user_id'] user_id, " +
                "data['origin_amount'] origin_amount, " +
                "data['coupon_reduce'] coupon_reduce, " +
                "data['final_amount'] final_amount, " +
                "data['create_time'] create_time, " +
                "data['update_time'] update_time, " +
                "pt, " +
                "ts " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_detail' " +
                "and `type`='insert'");

        tEnv.createTemporaryView("orderDetail",orderDetail);

        //tEnv.sqlQuery("select * from orderDetail").execute().print();

        //3order_info
        Table orderInfo = tEnv.sqlQuery("select " +
                "data['id'] id, " +
                "data['province_id'] province_id, " +
                "data['session_id'] session_id, " +
                "data['order_status'] order_status " +
                "from ods_db " +
                "where `database`='edu' " +
                "and `table`='order_info' " +
                "and `type`='insert'");

        tEnv.createTemporaryView("orderInfo",orderInfo);
       // tEnv.sqlQuery("select * from orderInfo").execute().print();

        //4.获取course_info
        tEnv.executeSql("create table course_info(" +
                "id bigint," +
                "subject_id bigint " +
                ")with(" +
                "'connector' = 'jdbc'," +
                "'url' = 'jdbc:mysql://hadoop102:3306/edu?useSSL=false'," +
                "'table-name' = 'course_info'," +
                "'username' = 'root'," +
                "'password' = '123321' " +
                ")");


        Table result = tEnv.sqlQuery("select " +
                " t1.id," +
                " t1.user_id," +
                " t1.course_id," +
                " t1.course_name," +
                " t3.subject_id," +
                " t1.order_id," +
                " t1.origin_amount," +
                " t1.coupon_reduce," +
                " t1.final_amount," +
                " t1.create_time," +
                " t1.update_time," +
                " t2.province_id," +
                " t2.session_id," +
                " t2.order_status," +
                " t1.ts," +
                "current_row_timestamp() row_op_ts " +
                " from orderDetail t1 " +
                " join orderInfo t2 on t1.order_id=t2.id " +
                " join course_info for system_time as of t1.pt t3 on cast(t1.course_id as bigint) = cast(t3.id as bigint) " +
                "");


        //写出到kafka
        tEnv.executeSql("create table  dwd_trade_order_detail( " +
                "id string, " +
                "user_id string, " +
                "course_id string, " +
                "course_name string, " +
                "subject_id bigint," +
                "order_id string, " +
                "origin_amount string, " +
                "coupon_reduce string, " +
                "final_amount string, " +
                "create_time string, " +
                "update_time string, " +
                "province_id string, " +
                "session_id string, " +
                "order_status string, " +
                "ts bigint, " +
                "row_op_ts timestamp_ltz(3) " +
                ")"+ SQLUtil.getKafkaSink(Constant.TOPIC_DWD_TRADE_ORDER_DETAIL)
        );

        result.executeInsert("dwd_trade_order_detail");



    }
}
